package stc.skymobi.netty.transport.endpoint;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;

import org.apache.commons.collections.Closure;
import org.jocean.util.ExceptionUtils;
import org.jocean.util.Visitor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.MBeanExporter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import stc.skymobi.netty.transport.Receiver;
import stc.skymobi.netty.transport.TransportUtils;

import javax.management.InstanceNotFoundException;
import javax.management.MBeanRegistrationException;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

public abstract class AbstractEndpoint<I> implements MutableEndpoint<I> {

    private static final Logger logger =
            LoggerFactory.getLogger(AbstractEndpoint.class);

    protected Closure messageClosure;
    protected Receiver<I> nextReceiver;
    protected Channel channel = null;

    protected BlockingQueue<I> pendings = null;

    protected MBeanExporter mbeanExport = null;

    protected String objectNamePrefix;

    protected ObjectName objectName = null;

    private final Map<UUID, Visitor<JSONObject>> _feedbacks = 
    		new ConcurrentHashMap<UUID, Visitor<JSONObject>>();
    /**
     * @return the objectNamePrefix
     */
    public String getObjectNamePrefix() {
        return objectNamePrefix;
    }

    /**
     * @param objectNamePrefix the objectNamePrefix to set
     */
    public void setObjectNamePrefix(String objectNamePrefix) {
        this.objectNamePrefix = objectNamePrefix;
    }

    /**
     * @return the mbeanExport
     */
    public MBeanExporter getMbeanExport() {
        return mbeanExport;
    }

    /**
     * @param mbeanExport the mbeanExport to set
     */
    public void setMbeanExport(MBeanExporter mbeanExport) {
        this.mbeanExport = mbeanExport;
    }

    @Override
    public void send(I bean) {
        pendings.add(bean);
    }

	//	action: invokeMBean -- invoke mbean's operation
    //	result: feedbackResult
    public void sendActionWithFeedback(final String action, final I bean, final Visitor<JSONObject> feedback ) {
    	//	record feedback info
    	//	and when message received via messageReceived
    	//	then feedback via feedback visitor
    	
    	//	gen JSONObject
    	final Object obj = JSON.toJSON(bean);
    	if ( !(obj instanceof JSONObject) ) {
    		//	return;
    		throw new RuntimeException("sendWithFeedback for invalid bean, bean to be send must be POJO");
    	}
    	
    	final JSONObject json = (JSONObject)obj;
    	json.put("_msgtype", action);
    	
    	final UUID transactionId = UUID.randomUUID();
    	json.put("_transactionId", transactionId.toString());
    	this._feedbacks.put(transactionId, feedback);
    	pendings.add((I)json);
    }
    
    private boolean tryMatchFeedbacks(final I msg) {
    	try {
	    	if ( msg instanceof JSONObject ) {
	    		final UUID transactionId = UUID.fromString( ((JSONObject) msg).getString("_transactionId") );
	    		if ( null != transactionId ) {
	    			final Visitor<JSONObject> feedback = this._feedbacks.remove(transactionId);
	    			if ( null != feedback ) {
	    				try {
							feedback.visit((JSONObject)msg);
						} catch (Exception e) {
							logger.error("exception when feedback for {}, detail: {}",
									msg, ExceptionUtils.exception2detail(e));
						}
	    				return	true;
	    			}
	    		}
	    	}
    	}
    	catch (Exception e) {
    		logger.error("exception when tryMatchFeedbacks, detail: {}", 
    				ExceptionUtils.exception2detail(e));
    	}
		return false;
	}

    public void messageReceived(ChannelHandlerContext ctx, I msg) {
    	if ( !tryMatchFeedbacks(msg) ) {
	        if (null != messageClosure) {
	            TransportUtils.attachSender(msg, this);
	            this.messageClosure.execute(msg);
	        }
	        if (null != nextReceiver) {
	            this.nextReceiver.messageReceived(ctx, msg);
	        }
    	}
    }

	public void setChannel(Channel channel) {
        this.channel = channel;
        if (null != objectName && null != mbeanExport) {
            try {
                mbeanExport.getServer().unregisterMBean(objectName);
            } catch (MBeanRegistrationException e) {
                logger.error("setChannel:", e);
                e.printStackTrace();
            } catch (InstanceNotFoundException e) {
                logger.error("setChannel:", e);
            }
        }
        if (null != channel && null != mbeanExport) {
            try {
                objectName = new ObjectName(objectNamePrefix + ",channel=" +
                        channel
                                .toString().replaceAll(":", "/")
                                .replaceAll(",", "")
                                .replaceAll("=>", "-")
                );
                mbeanExport.registerManagedResource(this, objectName);
            } catch (MalformedObjectNameException e) {
                logger.error("setChannel:", e);
            } catch (NullPointerException e) {
                logger.error("setChannel:", e);
            }
        }
    }

    /* (non-Javadoc)
      * @see com.skymobi.transport.Endpoint#stop()
      */
    public void stop() {
        if (null != objectName && null != mbeanExport) {
            try {
                mbeanExport.getServer().unregisterMBean(objectName);
            } catch (MBeanRegistrationException e) {
                logger.error("setIoSession:", e);
                e.printStackTrace();
            } catch (InstanceNotFoundException e) {
                logger.error("setIoSession:", e);
            }
        }
    }

    public void setMessageClosure(Closure closure) {
        this.messageClosure = closure;
    }

    public void setSendQueue(BlockingQueue<I> queue) {
        this.pendings = queue;
    }

    /**
     * @param nextReceiver the nextReceiver to set
     */
    public void setNextReceiver(Receiver<I> nextReceiver) {
        this.nextReceiver = nextReceiver;
    }

    public int getPendingCount() {
        if (null != this.pendings) {
            return this.pendings.size();
        } else {
            return -1;
        }
    }
}
